package com.bclz.environment;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.MRConfig;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/**
 * 二次排序
 * 输入各年的环境温度( year,month,date,temperature)，
 * 按照 year-month temperature1,temperature2...格式输出，并且按照年+月排序，如果相同则按温度排序
 *
 *          1. 构造中间键Environment,并且实现WritableComparable, map 完成后会按照实现方法compareTo规则排序；
 *          2. 构造分区实现类(继承Partitioner),实现map完成后的归约，指定map结果向哪个reduce端输出；
 *          3. 构造合并EnvironmentCombiner（继承WritableComparator），指定哪些数据作为reduce()方法的输入
 *
 *  hadoop jar algorithms-sparkOhadoop-1.0.0.jar com.bclz.environment.EnvironmentApplication  hdfs://master:9000/data/environment hdfs://master:9000/out
 *
 */


public class EnvironmentApplication extends Configured implements Tool {



    @Override
    public int run(String[] args) throws Exception {



//        Configuration conf=new Configuration();
        //也可不设置,本地运行默认
//        conf.set("fs.defaultFS","file:///");
//        conf.set(MRConfig.FRAMEWORK_NAME,"local");
        Configuration conf=getConf();
        conf.set("fs.defaultFS","hdfs://master:9000");
        conf.set(MRConfig.FRAMEWORK_NAME,"yarn");
        Job job=Job.getInstance(conf);
        job.setJarByClass(getClass());

        job.setPartitionerClass(EnvironmentPartitioner.class);
        job.setGroupingComparatorClass(EnvironmentCombiner.class);

        job.setMapperClass(EnvironmentMap.class);
        job.setMapOutputKeyClass(Environment.class);
        job.setMapOutputValueClass(IntWritable.class);

        job.setReducerClass(EnvironmentReduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        FileInputFormat.setInputPaths(job,new Path(args[0]));
        FileOutputFormat.setOutputPath(job,new Path(args[1]));


        boolean resultCode = job.waitForCompletion(true);

        return resultCode?0:1;
    }

    public static void main(String[] args) throws Exception {

        if(args.length!=2){
            throw new IllegalArgumentException("Usage: EnvironmentApplication [inputdir] [outputdir]");
        }

        int code = ToolRunner.run(new EnvironmentApplication(),args);
        System.exit(code);

    }
}
